package day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{Dependency, SparkConf, SparkContext}

/**
 * 2.7	RDD依赖关系
 */
object Dependencies {

}

/**
 * 2.7.1	查看血缘关系
 *
 * RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列
 * Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD
 * 的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息
 * 来重新运算和恢复丢失的数据分区。
 */
object Dependencies_view {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[String] = sc.textFile("input/a.txt")

    println(rdd.toDebugString)

    println("------------------------------")

    val newRdd: RDD[String] = rdd.flatMap(_.split(" "))

    println(newRdd.toDebugString)

    println("------------------------------")

    val mapRdd: RDD[(String, Int)] = newRdd.map((_, 1))

    println(mapRdd.toDebugString)

    println("------------------------------")

    val resRdd: RDD[(String, Int)] = mapRdd.reduceByKey(_ + _)

    println(resRdd.toDebugString)

    println("------------------------------")

    // 注意：圆括号中的数字表示RDD的并行度，也就是有几个分区
    //
    //    (2) ShuffledRDD[4] at reduceByKey at Dependencies.scala:48 []
    //    +-(2) MapPartitionsRDD[3] at map at Dependencies.scala:42 []
    //    |  MapPartitionsRDD[2] at flatMap at Dependencies.scala:36 []
    //    |  input/a.txt MapPartitionsRDD[1] at textFile at Dependencies.scala:30 []
    //    |  input/a.txt HadoopRDD[0] at textFile at Dependencies.scala:30 []

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.7.2	查看依赖关系
 *
 * 要想理解RDDS是如何工作的，最重要的就是理解Transformations。
 * RDD 之间的关系可以从两个维度来理解: 一个是 RDD 是从哪些 RDD 转换而来,
 * 也就是 RDD 的 parent RDD(s)是什么; 另一个就是 RDD 依赖于 parent RDD(s)
 * 的哪些 Partition(s). 这种关系就是 RDD 之间的依赖.
 *
 * RDD和它依赖的父RDD（s）的关系有两种不同的类型，
 * 即窄依赖（narrow dependency）和宽依赖（wide dependency）。
 *
 */
object Dependencies_view01 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val fileRdd: RDD[String] = sc.textFile("input/a.txt")

    val dependencies: Seq[Dependency[_]] = fileRdd.dependencies
    println(dependencies)
    println("------------------------------------")

    val wordRdd: RDD[String] = fileRdd.flatMap(_.split(" "))
    val dependencies1: Seq[Dependency[_]] = wordRdd.dependencies
    println(dependencies1)
    println("------------------------------------")

    val mapRdd: RDD[(String, Int)] = wordRdd.map((_, 1))
    val dependencies2: Seq[Dependency[_]] = mapRdd.dependencies
    println(dependencies2)
    println("------------------------------------")

    val resRdd: RDD[(String, Int)] = mapRdd.reduceByKey(_ + _)
    val dependencies3: Seq[Dependency[_]] = resRdd.dependencies
    println(dependencies3)
    println("------------------------------------")

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.7.3	窄依赖
 *
 * 窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition
 * 使用，窄依赖我们形象的比喻为独生子女。
 */
object Dependencies_narrow {

}

/**
 * 2.7.4	宽依赖
 * 宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖，
 * 会引起Shuffle，宽依赖我们形象的比喻为超生。
 *
 * 具有宽依赖的 transformations 包括: sort, reduceByKey, groupByKey, join
 * 和调用rePartition函数的任何操作.
 *
 * 宽依赖对 Spark 去评估一个 transformations 有更加重要的影响, 比如对性能的影响.
 *
 */
object Dependencies_shuffle {

}

/**
 * 2.7.5	Spark中的Job调度
 *
 * 一个Spark应用包含一个驱动进程(driver process，在这个进程中写Spark的逻辑代码)和
 * 多个执行器进程(executor process，跨越集群中的多个节点)。
 *
 * Spark 程序自己是运行在驱动节点, 然后发送指令到执行器节点。
 *
 * 一个Spark集群可以同时运行多个Spark应用, 这些应用是由集群管理器(cluster manager)来调度。
 * Spark应用可以并发的运行多个job， job对应着给定的应用内的在RDD上的每个action操作。
 *
 *
 *
 * 	Spark应用
 * 一个Spark应用可以包含多个Spark job, Spark job是在驱动程序中由SparkContext 来定义的。
 *
 * 当启动一个 SparkContext 的时候, 就开启了一个 Spark 应用。 一个驱动程序被启动了, 多个执行器
 * 在集群中的多个工作节点(worker nodes)也被启动了。 一个执行器就是一个 JVM, 一个执行器不能跨越
 * 多个节点, 但是一个节点可以包括多个执行器。
 *
 * 一个 RDD 会跨多个执行器被并行计算. 每个执行器可以有这个 RDD 的多个分区, 但是一个分区不能跨越
 * 多个执行器.
 *
 * 	Spark Job 的划分
 * 由于Spark的懒执行, 在驱动程序调用一个action之前, Spark 应用不会做任何事情，
 * 针对每个action，Spark 调度器就创建一个执行图(execution graph)和启动一个 Spark job。
 * 每个 job 由多个stages 组成, 这些 stages 就是实现最终的 RDD 所需的数据转换的步骤。
 * 一个宽依赖划分一个stage。每个 stage 由多个 tasks 来组成, 这些 tasks 就表示每个并行计算, 并且会在多个执行器上执行。
 *
 *
 */
object Job_scheduling {

}

/**
 *
 * 2.7.6	Stage任务划分（面试重点）
 *
 * 1）DAG有向无环图
 * DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。
 *
 * 原始的RDD通过一系列的转换就形成了DAG，根据RDD之间的依赖关系的不同将DAG划分成不同的Stage；
 *
 * 对于窄依赖，partition的转换处理在Stage中完成计算。
 * 对于宽依赖，由于有Shuffle的存在，只能在parent RDD处理完成后，才能开始接下来的计算，因此宽依赖是划分Stage的依据。
 *
 * 例如，DAG记录了RDD的转换过程和任务的阶段。
 *
 * 2）RDD任务切分中间分为：Application、Job、Stage和 Task
 *
 * ① Application：初始化一个SparkContext即生成一个Application；
 * ② Job：一个Action算子就会生成一个Job；
 * ③ Stage：Stage等于宽依赖的个数加1；
 * ④ Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。
 *
 * 注意：Application->Job->Stage->Task每一层都是1对n的关系。
 *
 */
object Stage_task_divide {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val dataRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 1, 2), 2)

    val resultRdd: RDD[(Int, Int)] = dataRdd.map((_, 1)).reduceByKey(_ + _)

    // 一个action算子就会生成一个job
    val tuples: Array[(Int, Int)] = resultRdd.collect()

    tuples.foreach(println)

    resultRdd.saveAsTextFile("output")

    // 休眠之后，查看 http://localhost:4040/jobs/
    Thread.sleep(90000)

    // 注意：如果存在shuffle过程，系统会自动进行缓存，UI界面显示skipped的部分

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.7.7	Stage任务划分源码分析
 */
object Stage_task_divide_source_code {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    val array: Array[Int] = rdd.collect()



    // 关闭连接
    sc.stop()
  }
}

